[GLUTEN-12306][FLINK][VL] Route print sink options to Velox via reflection (no file path)#12320
Merged
lgbo-ustc merged 7 commits intoJun 25, 2026
Merged
Conversation
ggjh-159
force-pushed
the
fix/print-sink-multi-parallelism
branch
from
914fab7 to
bc074f6
Compare
KevinyhZou
reviewed
Jun 23, 2026
| Field idField = writer.getClass().getDeclaredField("sinkIdentifier"); | ||
| idField.setAccessible(true); | ||
| Field stdErrField = writer.getClass().getDeclaredField("target"); | ||
| stdErrField.setAccessible(true); |
Contributor
There was a problem hiding this comment.
Use ReflectUtils.getObjectField() should be simpler
| // Flink 1.19.x field names: sinkIdentifier (print-identifier), target (standard-error, true = | ||
| // stderr). | ||
| // Package-private for direct unit testing. | ||
| static String[] extractPrintOptions(Transformation<RowData> transformation) { |
Contributor
There was a problem hiding this comment.
the readability of String[] is bad， may we can use class PringOptions instead?
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| class PrintSinkFactoryTest { |
Contributor
There was a problem hiding this comment.
GlutenStreamingTestBase.runAndCheck() still read taskmanager.out, please check whether it should be fixed together?
ggjh-159
force-pushed
the
fix/print-sink-multi-parallelism
branch
from
8c5eba9 to
adcbeaf
Compare
Contributor
Author
|
cc @KevinyhZou |
Contributor
Author
|
@lgbo-ustc Would you mind re-triggering the CI for this PR? I've pushed a new commit and would like to confirm the Flink tests. |
Contributor
Contributor
Author
… captured fd=1 Reverts b561569. That commit removed File.mkdirs() under the mistaken assumption that CoreOptions.FLINK_LOG_DIR would create the dir. Under the dup2(fd=1) capture mechanism, Flink never creates that dir, so C_LIB.open() returned -1 and ScalarFunctionsTest/ScanTest failed with FlinkRuntimeException at runAndCheck:147. Restoring the 59015d1 state where mkdirs() is the dir creator and FLINK_LOG_DIR is intentionally absent (dup2 makes it redundant).
Contributor
Author
|
@lgbo-ustc Sorry, I accidentally reverted a previous unit test fix. It's now restored. Full local run of the gluten-flink ut module: 
Could you please rebuild the CI?
|
lgbo-ustc
approved these changes
Jun 25, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
fix: #12306
related prs: bigo-sg/velox#49 bigo-sg/velox4j#39
What changes are proposed in this pull request?
This PR consumes the companion velox PR (which now writes to
std::cout/std::cerr):Configuration/CoreOptions/ConfigConstantsimports.PrintSinkFactory.extractPrintOptionsreflects into theRowDataPrintFunction→PrintSinkOutputWriterto read the user-suppliedsinkIdentifier(print-identifier) andtarget(true = stderr) fields, and passes them through toPrintTableHandle(tableName, inputColumns, printIdentifier, isStdErr).N>prefix fromparallelism/task_indexsession properties and serializes concurrent subtask writes with a process-wide mutex.How was this patch tested?
parallelism.default = 2, nexmarkevents.num = 10000, tps = 2000, query q0: every bid row reachestaskmanager.out, each line carries the1>/2>subtask prefix, no truncation across subtasks.